package com.rocketmq.temp2.producer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

@RestController
public class ProducerController {

    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    public ProducerController(RocketMQTemplate rocketMQTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
    }

    @GetMapping(value = "/sendMember")
    public String sendMember() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Long orderId = System.currentTimeMillis();

        String insertSqlMsg = getSqlMessage("insert", orderId);
        String updateSqlMsg = getSqlMessage("update", orderId);
        String deleteSqlMsg = getSqlMessage("delete", orderId);

        Message insertMsg = new Message("sm-platform-topic", insertSqlMsg.getBytes());
        Message updateMsg = new Message("sm-platform-topic", updateSqlMsg.getBytes());
        Message deleteMsg = new Message("sm-platform-topic", deleteSqlMsg.getBytes());

        DefaultMQProducer producer = rocketMQTemplate.getProducer();
        producer.send(insertMsg, new RocketMessageQueueSelector(), orderId);
        producer.send(updateMsg, new RocketMessageQueueSelector(), orderId);
        producer.send(deleteMsg, new RocketMessageQueueSelector(), orderId);

        return "success";
    }

    private String getSqlMessage(String type, Long orderId) {
        ObjectNode jsonNodes = new ObjectMapper().getNodeFactory().objectNode();
        jsonNodes.put("type", type);
        jsonNodes.put("orderId", orderId);
        return jsonNodes.toString();
    }

    static class RocketMessageQueueSelector implements MessageQueueSelector {
        /**
         * 自定义消息队列选举机制
         *
         * @param mqList 集群队列总数
         * @param msg    消息内容
         * @param msgId  传递的参数(该示例中指orderId)
         * @return 返回被选举的队列
         */
        @Override
        public MessageQueue select(List<MessageQueue> mqList, Message msg, Object msgId) {
            //Long argOrderId = Long.parseLong(String.valueOf(msgId));
            //Long index = argOrderId % mqList.size();
            //return mqList.get(index.intValue());
            // 该消息存放到 索引为0 的队列中
            return mqList.get(0);
        }
    }
}
